package com.rainwen.rabbitmq.server.service;

import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Map;
import java.util.Properties;
import java.util.Set;

/**
 * Created by rain.wen on 2017/7/20.
 */
@Service
public class MessageServiceImpl implements MessageService {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private RabbitAdmin rabbitAdmin;

    /**
     * 发送消息
     *
     * @param exchange   交换区
     * @param routingKey routingKey值
     * @param object     对象
     * @param header     头属性
     */
    @Override
    public void send(String exchange, String routingKey, Object object, Map header) {
        send(exchange,routingKey,object,header,0L);
    }

    /**
     * 发送消息
     *
     * @param exchange   交换区
     * @param routingKey routingKey值
     * @param object     对象
     * @param header     头属性
     * @param expires    过期时间 毫秒
     */
    @Override
    public void send(String exchange, String routingKey, Object object, final Map header, final long expires) {
        rabbitTemplate.convertAndSend(exchange, routingKey, object, new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                if (expires > 0L) {
                    message.getMessageProperties().setExpiration(String.valueOf(expires));
                }
                if (header != null) {
                    Set<String> set = header.keySet();
                    for (String key : set) {
                        message.getMessageProperties().setHeader(key, header.get(key));
                    }
                }

                return message;
            }
        });
    }

    /**
     * 创建并绑定一个队列到一个交换,并且制定对应的路由键
     *
     * @param exchange
     * @param queueName
     * @param routingKey
     */
    @Override
    public void declareExchangeAndQueue(String exchange, String queueName, String routingKey) {
        //创建一个队列
        Queue queue = new Queue(queueName);
        rabbitAdmin.declareQueue(queue);
        //创建交换
        TopicExchange topicExchange = new TopicExchange(exchange);
        rabbitAdmin.declareExchange(topicExchange);
        //绑定队列到交换
        rabbitAdmin.declareBinding(BindingBuilder.bind(queue).to(topicExchange).with(routingKey));
    }

    /**
     * 删除一个交换
     *
     * @param exchangeName
     * @return
     */
    @Override
    public boolean deleteExchange(String exchangeName) {
        return rabbitAdmin.deleteExchange(exchangeName);
    }

    /**
     * 删除一个队列
     *
     * @param queueName
     * @param unused
     * @param empty
     */
    @Override
    public void deleteQueue(String queueName, boolean unused, boolean empty) {
        rabbitAdmin.deleteQueue(queueName,unused,empty);
    }

    /**
     * 清空一个队列
     *
     * @param queueName
     * @param noWait
     */
    @Override
    public void purgeQueue(String queueName, boolean noWait) {
        rabbitAdmin.purgeQueue(queueName,noWait);
    }

    /**
     * 获取一个队列的属性
     *
     * @param queueName
     * @return
     */
    @Override
    public Properties getQueueProperties(String queueName) {
        return rabbitAdmin.getQueueProperties(queueName);
    }
}
